package jmeter.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

public class EchoNIOServer {
	protected static final Logger LOCAL_LOG = LogManager
			.getLogger(LogManager.ROOT_LOGGER_NAME);

	private final ExecutorService executorServiceWrite = Executors
			.newFixedThreadPool(7);

	private final ExecutorService executorServiceRead = Executors
			.newFixedThreadPool(4);

	private final BlockingQueue<ChannelData> receivedSockets = new LinkedBlockingQueue<ChannelData>();

	private ThreadReader reader;
	private ThreadWriter writer;
	private Selector selectorWriter;

	private static Selector selectorReader;
	
	private static final EchoNIOServer nioServer = new EchoNIOServer();

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		try {
			

			nioServer.selectorReader = Selector.open();
			nioServer.selectorWriter = Selector.open();

			nioServer.writer = nioServer.new ThreadWriter(
					nioServer.selectorWriter, nioServer.receivedSockets);
			nioServer.reader = nioServer.new ThreadReader(selectorReader,
					nioServer.receivedSockets);

			int port = 12345;
			if (args.length > 0) {
				port = Integer.parseInt(args[0]);
			}
			ServerSocketChannel serverChannel = ServerSocketChannel.open();
			ServerSocket serverSocket = serverChannel.socket();
			final Selector selectorAccepter = Selector.open();
			serverSocket.bind(new InetSocketAddress(port));
			serverChannel.configureBlocking(false);
			serverChannel.register(selectorAccepter, SelectionKey.OP_ACCEPT);

			new Thread(new Runnable() {

				@Override
				public void run() {
					while (true)
						try {
							// synchronized (selectorReader) {
							// }

							if (0 == selectorReader.selectNow())
								continue;

							Iterator<SelectionKey> it = selectorReader
									.selectedKeys().iterator();
							while (it.hasNext()) {
								SelectionKey key = (SelectionKey) it.next();
								if (key.isReadable()) {
									SocketChannel socketChannel = (SocketChannel) key
											.channel();
									ByteBuffer buffer = ByteBuffer
											.wrap(new byte[256]);
									socketChannel.read(buffer);
									buffer.flip();
									nioServer.receivedSockets.offer(
											nioServer.new ChannelData(key,
													buffer.array()));
									// registerChannel(nioServer.selectorWriter,
									// key.channel(),
									// SelectionKey.OP_WRITE);
									nioServer.executorServiceWrite
											.submit(nioServer.writer);
									buffer.clear();
									key.cancel();
								}
							}
						} catch (IOException e) {
							LOCAL_LOG.error(e.getMessage(), e);
						} catch (Throwable t) {
							LOCAL_LOG.fatal(t.getMessage(), t);
						}
				}
			}).start();
			new Thread(new Runnable() {

				@Override
				public void run() {
					while (true) {
						try {
							int n = selectorAccepter.select();
							if (n == 0) {
								continue;
							}
							Iterator<SelectionKey> it = selectorAccepter
									.selectedKeys().iterator();
							while (it.hasNext()) {
								SelectionKey key = (SelectionKey) it.next();
								if (key.isAcceptable()) {
									ServerSocketChannel server = (ServerSocketChannel) key
											.channel();
									SocketChannel channel = server.accept();
									// synchronized (selectorReader) {
									// selectorReader.wakeup();
									nioServer.registerChannel(selectorReader,
											channel, SelectionKey.OP_READ);
									// }
								}
								it.remove();
							}
						} catch (IOException e) {
							LOCAL_LOG.error(e.getMessage(), e);
						} catch (Throwable t) {
							LOCAL_LOG.fatal(t.getMessage(), t);
						}
					}

				}
			}).start();
		} catch (IOException exception) {
			LOCAL_LOG.error(exception.getMessage(), exception);
		} finally {

		}
	}

	protected void registerChannel(Selector selector,
			SelectableChannel channel, int ops) throws IOException {
		if (channel == null) {
			return; // could happen
		}
		channel.configureBlocking(false);
		channel.register(selector, ops);
	}

	public ExecutorService getExecutorServiceWrite() {
		return executorServiceWrite;
	}

	class ChannelData {
		private final SelectionKey selectionKey;
		private final byte[] bs;

		public ChannelData(final SelectionKey selectionKey, final byte[] bs) {
			this.selectionKey = selectionKey;
			this.bs = bs;
		}

		public SelectionKey getSelectionKey() {
			return selectionKey;
		}

		public byte[] getBs() {
			return bs;
		}
	}

	private class ThreadReader implements Callable<ChannelData> {
		private final Selector selectorReader;
		private final BlockingQueue<ChannelData> receivedSockets;

		public ThreadReader(final Selector selectorReader,
				final BlockingQueue<ChannelData> receivedSockets) {
			this.selectorReader = selectorReader;
			this.receivedSockets = receivedSockets;
		}

		@Override
		public ChannelData call() throws Exception {
			if (0 < selectorReader.select()) {
				Iterator<SelectionKey> it = selectorReader.selectedKeys()
						.iterator();
				while (it.hasNext()) {
					SelectionKey key = (SelectionKey) it.next();

					it.remove();
				}
			}
			return null;
		}

	}

	private class ThreadWriter implements Callable<ChannelData> {
		private final Selector selectorWriter;
		private final BlockingQueue<ChannelData> receivedSockets;
		private final ChannelData[] datas = new ChannelData[0];

		public ThreadWriter(final Selector selectorWriter,
				final BlockingQueue<ChannelData> receivedSockets) {
			this.selectorWriter = selectorWriter;
			this.receivedSockets = receivedSockets;
		}

		@Override
		public ChannelData call() throws Exception {
			ChannelData channelData = null;
			try {
				channelData = this.receivedSockets.poll();
				if (channelData != null) {
					ByteBuffer byteBuffer = ByteBuffer
							.wrap(channelData.getBs());
					((SocketChannel) channelData.getSelectionKey().channel())
							.write(byteBuffer);
					for (ChannelData c : this.receivedSockets.toArray(datas)) {
						if (c != null)
							byteBuffer.rewind();
						synchronized (((SocketChannel) channelData
								.getSelectionKey().channel())) {
							if (((SocketChannel) channelData.getSelectionKey()
									.channel()).isConnected())
								((SocketChannel) channelData.getSelectionKey()
										.channel()).write(byteBuffer);
						}
					}
				}
			} catch (IOException e) {
				LOCAL_LOG.error(e.getMessage(), e);
			} finally {
				if (channelData != null)
					synchronized (channelData.getSelectionKey().channel()) {
						channelData.getSelectionKey().channel().close();
					}
			}
			return null;
		}
	}
}
